package com.firewolf.rocketmq;

import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
import org.apache.rocketmq.client.consumer.DefaultMQPullConsumer;
import org.apache.rocketmq.client.consumer.PullResult;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.message.MessageQueue;

/**
 * 作者：刘兴 时间:2019-07-15
 **/
public class PullConsumer {

  private static final Map<MessageQueue, Long> OFFSE_TABLE = new HashMap<MessageQueue, Long>();

  public static void main(String[] args) throws MQClientException {
    DefaultMQPullConsumer consumer = new DefaultMQPullConsumer("consumer-group1");

    consumer.setNamesrvAddr("10.211.55.6:9876");
    consumer.start();

    Set<MessageQueue> mqs = consumer.fetchSubscribeMessageQueues("mytopic");
    for (MessageQueue mq : mqs) {
      System.out.printf("Consume from the queue: %s%n", mq);
      SINGLE_MQ:
      while (true) { //先获取完一个Message Queue中的消息
        try {
          PullResult pullResult =
              consumer.pullBlockIfNotFound(mq, null, getMessageQueueOffset(mq), 32);
          System.out.printf("%s%n", pullResult);
          //更新offset
          putMessageQueueOffset(mq, pullResult.getNextBeginOffset());
          switch (pullResult.getPullStatus()) {
            case FOUND:
              dealPullResult(pullResult);
              break;
            case NO_MATCHED_MSG:
              break;
            case NO_NEW_MSG:
              break SINGLE_MQ;
            case OFFSET_ILLEGAL:
              break;
            default:
              break;
          }
        } catch (Exception e) {
          e.printStackTrace();
        }
      }
    }

    consumer.shutdown();
  }

  private static void dealPullResult(PullResult pullResult) {

    List<MessageExt> msgFoundList = pullResult.getMsgFoundList();
    for (MessageExt messageExt : msgFoundList) {
      System.out.println(new String(messageExt.getBody()));
    }

  }

  //获取offset
  private static long getMessageQueueOffset(MessageQueue mq) {
    Long offset = OFFSE_TABLE.get(mq);
    if (offset != null) {
      return offset;
    }
    return 0;
  }


  //更新offset
  private static void putMessageQueueOffset(MessageQueue mq, long offset) {
    OFFSE_TABLE.put(mq, offset);
  }
}
